package com.atguigu.edu.app.dwd.db;

import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;

/**
 * @author zzw
 * @create 2022/9/3 16:26
 */
public class DwdTradeOrderPreProcess {
    public static void main(String[] args) {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        env.setStateBackend(new HashMapStateBackend());
        */

        // 设置时区的方法
        tableEnv.getConfig().setLocalTimeZone(ZoneId.of("GMT+8"));

        // TODO 3 读取kafka topic_db  和topic_log主题数据
        String topicName = "topic_db";
        String logTopic = "topic_log";
        String groupId = "dwd_trade_order_pre_process";
        tableEnv.executeSql("create table topic_db(\n" +
                "    `database` String,\n" +
                "    `table` String,\n" +
                "    `type` String,\n" +
                "    `ts` String,\n" +
                "    `xid` String,\n" +
                "    `commit` String,\n" +
                "    `data` map<string,string>,\n" +
                "    `old` map<string,string>,\n" +
                "    `pt` AS PROCTIME()\n" +
                ") " + KafkaUtil.getKafkaDDL(topicName,groupId));

        tableEnv.executeSql("create table topic_log(\n" +
                "`common` map<String,String>,\n" +
                "`pt` AS PROCTIME()\n" +
                ")" + KafkaUtil.getKafkaDDL(logTopic,groupId));

        // TODO 4 过滤出session数据
        Table sessionTable = tableEnv.sqlQuery("select\n" +
                "`common`['sc'] sc,\n" +
                "`common`['sid'] sid,\n" +
                "`common`['uid'] uid\n" +
                "from\n" +
                "topic_log");
        tableEnv.createTemporaryView("session_info",sessionTable);

        // TODO 4 过滤出订单详情表数据

        Table orderDetailTable = tableEnv.sqlQuery("select \n" +
                "`data`['id'] id , \n" +
                "`data`['course_id'] course_id , \n" +
                "`data`['course_name'] course_name , \n" +
                "`data`['order_id'] order_id,  \n" +
                "`data`['user_id'] user_id,  \n" +
                "`data`['origin_amount'] origin_amount,  \n" +
                "`data`['coupon_reduce'] coupon_reduce , \n" +
                "`data`['final_amount'] final_amount,  \n" +
                "`data`['create_time'] create_time,  \n" +
                "`data`['update_time'] update_time  \n" +
                "from topic_db \n" +
                "where `table` = 'order_detail'\n" +
                "and `type` = 'insert'");
        tableEnv.createTemporaryView("order_detail",orderDetailTable);

        // TODO 5 过滤出订单表数据
        Table orderInfoTable = tableEnv.sqlQuery("select  \n" +
                "`data`['id'] id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['order_status'] order_status,\n" +
                "`data`['out_trade_no'] out_trade_no,\n" +
                "`data`['trade_body'] trade_body,\n" +
                "`data`['session_id'] session_id,\n" +
                "`data`['province_id'] province_id,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['expire_time'] expire_time,\n" +
                "`data`['update_time'] update_time,\n" +
                "`type`,\n" +
                "`old`,\n" +
                "`ts`\n" +
                "from topic_db\n" +
                "where `table` = 'order_info'\n" +
                "and (`type` = 'insert' or `type` = 'update')");
        tableEnv.createTemporaryView("order_info",orderInfoTable);

        // TODO 6 join3张表
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "od.id,\n" +
                "od.course_id,\n" +
                "od.course_name,\n" +
                "od.order_id,\n" +
                "od.user_id,\n" +
                "od.origin_amount,\n" +
                "od.coupon_reduce,\n" +
                "od.final_amount,\n" +
                "od.create_time,\n" +
                "od.update_time,\n" +
                "oi.order_status,\n" +
                "oi.out_trade_no,\n" +
                "oi.trade_body,\n" +
                "si.sc,\n" +
                "oi.session_id,\n" +
                "oi.province_id,\n" +
                "oi.`type`,\n" +
                "oi.`old`,\n" +
                "oi.`ts`\n" +
                "from\n" +
                "order_detail od\n" +
                "join order_info oi\n" +
                "on od.order_id = oi.id\n" +
                "join session_info si\n" +
                "on (oi.session_id = si.sid and oi.user_id = si.uid)");
        tableEnv.createTemporaryView("result_table",resultTable);

        // TODO 7 写出到新主题 dwd_trade_order_pre_process
        tableEnv.executeSql("create table kafka_sink(\n" +
                "id String,\n" +
                "course_id String,\n" +
                "course_name String,\n" +
                "order_id String,\n" +
                "user_id String,\n" +
                "origin_amount String,\n" +
                "coupon_reduce String,\n" +
                "final_amount String,\n" +
                "create_time String,\n" +
                "update_time String,\n" +
                "order_status String,\n" +
                "out_trade_no String,\n" +
                "trade_body String,\n" +
                "sc String,\n" +
                "session_id String,\n" +
                "province_id String,\n" +
                "`type` String,\n" +
                "`old` map<String,string>,\n" +
                "`ts` String\n" +
                ")\n" + KafkaUtil.getKafkaSinkDDL("dwd_trade_order_pre_process"));

        tableEnv.executeSql("insert into kafka_sink select * from result_table");

    }
}
